package com.fiberhome.boot.conf.rabbitmq;

import com.fiberhome.boot.demo.rabbitmq.dao.PersonDao;
import com.fiberhome.boot.demo.rabbitmq.domain.Person;
import com.fiberhome.boot.utils.ObjectAndByte;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;
import java.io.IOException;


/**
 * 监听的业务类，实现接口MessageListener
 * Created by ywhu.
 */

@RabbitListener(containerFactory = "rabbitListenerContainer", queues = AmqpConfig.QUEUE_NAME)
public class Receiver implements ChannelAwareMessageListener {
    public Logger logger = LoggerFactory.getLogger(this.getClass());


    @Resource
    private PersonDao personDao;

    @Autowired
    private RabbitTemplate template;
    
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {

        try {

            logger.info("##### = {}", new String(message.getBody()));
            Person person =  (Person)ObjectAndByte.toObject(message.getBody());
            //模拟耗时操作，将会导致消息生产者接受不到消费信息
            //TimeUnit.SECONDS.sleep(10);
            boolean result =  personDao.save(person) !=null;
            if (!result) {
                logger.error("消息消费失败");
            } else {
                logger.info("消息消费成功");
            }
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);//手动消息应答
        } catch (IOException e) {
            logger.error(e.getMessage(), e);
//            channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);//true 重新放入队列
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);//对于处理不了的异常消息
            Person person = (Person) ObjectAndByte.toObject(message.getBody());
            //发送到失败队列
            template.convertAndSend(AmqpConfig.EXCHANGE, AmqpConfig.ROUTINGKEY_FAIL, person);
        }
    }
}
